package cn.doitedu.rtdw.data_etl;

import cn.doitedu.rtdw.udf.StrArrayLen;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: deep as the sea
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/4/10
 * @Desc: 学大数据，到多易教育
 *  推荐栏位事件实时分析etl任务
 **/
public class Etl06_RecommendEventsAnalyse {

    public static void main(String[] args) {
        // 编程入口
        //StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",8081);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);


        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.setParallelism(1);
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 创建kafka连接器表，来映射kafka中的dwd层的用户行为公共维度宽表
        tenv.executeSql(
                "  CREATE TABLE dwd_kafka (                                   "
                        + "     user_id           BIGINT,                     "
                        + "     event_id          string,                     "
                        + "     event_time        bigint,                     "
                        + "     device_type       string,                     "
                        + "     page_type         string,                     "
                        + "     properties        map<string,string>,         "
                        + "     pt AS proctime() ,                            "
                        + "     rt AS to_timestamp_ltz(event_time,3),         "
                        + "     WATERMARK FOR rt AS rt - interval '0' second  "
                        + " ) WITH (                                          "
                        + "  'connector' = 'kafka',                           "
                        + "  'topic' = 'dwd-events-detail',                   "
                        + "  'properties.bootstrap.servers' = 'doitedu:9092', "
                        + "  'properties.group.id' = 'testGroup',             "
                        + "  'scan.startup.mode' = 'latest-offset',           "
                        + "  'value.format'='json',                           "
                        + "  'value.json.fail-on-missing-field'='false',      "
                        + "  'value.fields-include' = 'EXCEPT_KEY')           ");

        // 创建hbase连接器表，映射hbase中的视频信息维表
        tenv.executeSql("CREATE TABLE recommend_hbase( " +
                " rec_region_id STRING, " +
                " f ROW<rec_name STRING, rec_type STRING, rec_suan STRING >, " +
                " PRIMARY KEY (rec_region_id) NOT ENFORCED " +
                ") WITH (                                                 " +
                " 'connector' = 'hbase-2.2',                              " +
                " 'table-name' = 'dim_rec_info',                          " +
                " 'zookeeper.quorum' = 'doitedu:2181',                    " +
                " 'lookup.cache' = 'PARTIAL',                             " +
                " 'lookup.partial-cache.max-rows' = '10000',              " +
                " 'lookup.partial-cache.expire-after-access' = '6000 s'    " +
                ")");

        /**
         * 数据扁平化，lookup 维表关联
         */
        tenv.executeSql(
                " CREATE TEMPORARY VIEW tmp_view AS                                   "
                        +" WITH tmp as (                                              "
                        +" SELECT                                                     "
                        +"  user_id                                                   "
                        +"  ,event_id                                                 "
                        +"  ,event_time                                               "
                        +"  ,device_type                                              "
                        +"  ,properties                                               "
                        +"  ,properties['rec_region_id']  as rec_region_id            "
                        +"  ,properties['show_items']  as show_items                  "
                        +"  ,properties['item_id']  as click_item_id                  "
                        +"  ,properties['rec_tracking_id'] AS rec_tracking_id         "
                        +"  ,page_type                                                "
                        +"  ,pt                                                       "
                        +"  ,rt                                                       "
                        +" FROM  dwd_kafka                                            "
                        +" WHERE event_id in ('rec_show','rec_click')                 "
                        +" )                                                          "
                        +" SELECT                                                     "
                        +"     tmp.user_id                                            "
                        +"     ,tmp.device_type                                       "
                        +"     ,tmp.page_type                                         "
                        +"     ,tmp.rec_region_id                                     "
                        +"     ,tmp.rec_tracking_id                                   "
                        +"     ,tmp.show_items                                        "
                        +"     ,tmp.click_item_id                                     "
                        +"     ,tmp.rt                                                "
                        +"     ,h.rec_type AS rec_region_type                         "
                        +"     ,h.rec_name AS rec_region_name                         "
                        +"     ,h.rec_suan AS rec_math_model                          "
                        +" FROM tmp                                                   "
                        +" left join recommend_hbase FOR SYSTEM_TIME AS OF tmp.pt AS h "
                        +" ON tmp.properties['rec_region_id'] = h.rec_region_id        "
        );


        /**
         *  创建doris连接器表，映射轻度聚合的目标物理表
         */
        tenv.executeSql(
                " create table rec_doris (            "
                        + "     user_id          BIGINT,          "
                        + "     device_type      VARCHAR(20), "
                        + "     page_type        VARCHAR(20),  "
                        + "     rec_region_id    VARCHAR(60),  "
                        + "     rec_tracking_id    VARCHAR(60),  "
                        + "     rec_region_type  VARCHAR(60),  "
                        + "     rec_region_name  VARCHAR(60),  "
                        + "     rec_math_model        VARCHAR(60),  "
                        + "     show_item_cnt        BIGINT,     "
                        + "     click_item_cnt      BIGINT      "
                        + " ) WITH (                               "
                        + "    'connector' = 'doris',              "
                        + "    'fenodes' = 'doitedu:8030',         "
                        + "    'table.identifier' = 'dws.recommend_ana_agg',  "
                        + "    'username' = 'root',                "
                        + "    'password' = '',                    "
                        + "    'sink.label-prefix' = 'doris_tl" + System.currentTimeMillis() + "')");

        /**
         * 5分钟滚动时间窗口
         * 按照相同推荐跟踪id聚合
         */
        tenv.createTemporaryFunction("str_arr_len", StrArrayLen.class);
        tenv.executeSql(
                " INSERT INTO rec_doris                                                "
               // " create temporary view result_view as                                  "
                        +" SELECT                                                       "
                        +"    user_id,                                                  "
                        +"    device_type,                                              "
                        +"    page_type,                                                "
                        +"    rec_region_id,                                            "
                        +"    rec_tracking_id,                                          "
                        +"    rec_region_type,                                          "
                        +"    rec_region_name,                                          "
                        +"    rec_math_model,                                           "
                        +"    sum(str_arr_len(show_items,',')) as show_item_cnt,        "
                        +"    count(click_item_id) as click_item_cnt                    "
                        +" FROM TABLE(                                                  "
                        +"  TUMBLE(TABLE tmp_view,DESCRIPTOR(rt),INTERVAL '1' SECOND)   "
                        +" )                                                            "
                        +" GROUP BY                                                     "
                        +"    window_start,                                             "
                        +"    window_end,                                               "
                        +"    user_id,                                                  "
                        +"    device_type,                                              "
                        +"    page_type,                                              "
                        +"    rec_region_id,                                            "
                        +"    rec_tracking_id,                                          "
                        +"    rec_region_type,                                          "
                        +"    rec_region_name,                                          "
                        +"    rec_math_model                                            "


        );


        //tenv.executeSql("select * from result_view").print();

    }
}
